package com.hujf.flink.tableSql;

import com.hujf.flink.ch05.ClickSource;
import com.hujf.flink.ch05.Event;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;

import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author hujf
 * @Date 2022/3/24 17:13
 * @Description TODO
 */
public class SimpleTableExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);
        environment.setRuntimeMode(RuntimeExecutionMode.BATCH);
        SingleOutputStreamOperator<Event> eventDataStreamSource =  environment.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.getTimestamp();
                            }
                        }));

        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(environment);

        Table table = tableEnvironment.fromDataStream(eventDataStreamSource,$("user"), $("url"),$("timestamp"));

        Table result = tableEnvironment.sqlQuery("select* from " + table + "  where user = 'Bob'");
        //通过对表操作过滤
        Table where = table.select($("user"), $("url"),$("timestamp")).where($("user").isEqual("Bob"));


        tableEnvironment.toDataStream(result).print("result1->");
        tableEnvironment.toDataStream(where).print("result2->");


        tableEnvironment.createTemporaryView("clickTable",table);
        Table table1 = tableEnvironment.sqlQuery("select user, count(url) as cnt from clickTable group by user");

        tableEnvironment.toChangelogStream(table1).print();

        environment.execute();
    }
}
